package com.bite.messageQueue.mqServer.datacenter;

import com.bite.messageQueue.common.MqException;
import com.bite.messageQueue.mqServer.core.Binding;
import com.bite.messageQueue.mqServer.core.Exchange;
import com.bite.messageQueue.mqServer.core.MSGQueue;
import com.bite.messageQueue.mqServer.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用这个类来管理内存中的数据
 * 可能会在多线程下进行，要注意多线程问题
 */
@Slf4j
public class MemoryDataCenter {
    //key是exchangeName，value是Exchange
    private ConcurrentHashMap<String, Exchange> exchangesMap = new ConcurrentHashMap<>();
    //key是queueName，value是MSGQueue
    private ConcurrentHashMap<String, MSGQueue> queuesMap = new ConcurrentHashMap<>();
    //外部key是exchangeName，value是ConcurrentHashMap，内部中key是queueName，value是Binding
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key是messageId，value是Message
    private ConcurrentHashMap<String, Message> messagesMap = new ConcurrentHashMap<>();
    //key是queueName，value是queue中的消息列表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //key是queueName，value是queue中等待应答的消息列表
    private ConcurrentHashMap<String, ConcurrentHashMap<String,Message>> queueMessageMapWaitACK = new ConcurrentHashMap<>();

    /**
     * 交换机操作
     */
    public void insertExchange(Exchange exchange){
        exchangesMap.put(exchange.getName(),exchange);
    }

    public void deleteExchange(String exchangeName){
        exchangesMap.remove(exchangeName);
    }

    public Exchange getExchange(String exchangeName){
        return exchangesMap.get(exchangeName);
    }

    /**
     * 队列操作
     */
    public void insertQueue(MSGQueue queue){
        queuesMap.put(queue.getName(),queue);
    }

    public void deleteQueue(String queueName){
        queuesMap.remove(queueName);
    }

    public MSGQueue getQueue(String queueName){
        return queuesMap.get(queueName);
    }

    /**
     * 绑定操作
     */
    public void insertBinding(Binding binding) throws MqException {
        /*ConcurrentHashMap<String,Binding> bindingsMap = this.bindingsMap.get(binding.getExchangeName());
        if(bindingsMap == null){
            bindingsMap = new ConcurrentHashMap<>();
            this.bindingsMap.put(binding.getExchangeName(),bindingsMap);
        }*/
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());

        synchronized(bindingMap){
            //再根据queueName判断是否已经存在，如果存在则抛出异常
            if(bindingMap.get(binding.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("[MemoryDataCenter] 绑定删除成功! exchangeName={}, queueName={}", binding.getExchangeName(), binding.getQueueName());
    }

    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    /**
     * 消息操作
     */
    public void addMessage(Message message){
        messagesMap.put(message.getMessageID(),message);
        log.info("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageID());
    }

    public void removeMessage(String messageId){
        messagesMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

    public Message getMessage(String messageId){
        return messagesMap.get(messageId);
    }

    /**
     * 队列中的消息操作
     */
    public void sendMessageToQueue(MSGQueue queue,Message message){
        LinkedList<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized(messageList){
            messageList.add(message);
        }
        //向消息集合中添加消息
        addMessage(message);
        log.info("[MemoryDataCenter] 消息被投递到队列中! messageId=" + message.getMessageID());
    }

    public Message pollMessage(String queueName){
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        if(messageList == null){
            log.info("[MemoryDataCenter] 队列中没有消息! queueName=" + queueName);
            return null;
        }
        synchronized (messageList){
            if(messageList.size() == 0){
                log.info("[MemoryDataCenter] 队列长度为0! queueName=" + queueName);
                return null;
            }
            log.info("[MemoryDataCenter] 消息被消费! messageId=" + messageList.get(0).getMessageID());
            return messageList.remove(0);
        }
    }

    public int getMessageSize(String queueName){
        LinkedList<Message> messageList = queueMessageMap.get(queueName);
        if(messageList == null){
            return 0;
        }
        synchronized (messageList){
            return messageList.size();
        }
    }

    /**
     * 处理未确认的消息
     */
    public void addMessageWaitACK(String queueName, Message message){
        ConcurrentHashMap<String, Message> messageWaitACKMap = queueMessageMapWaitACK.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        messageWaitACKMap.put(message.getMessageID(),message);
        log.info("[MemoryDataCenter] 未确认消息进入确认队列! queueName={}, messageId={}", queueName, message.getMessageID());
    }

    public void removeMessageWaitACK(String queueName, String messageId) throws MqException {
        ConcurrentHashMap<String, Message> messageWaitACKMap = queueMessageMapWaitACK.get(queueName);
        if(messageWaitACKMap == null){
            return;
        }
        messageWaitACKMap.remove(messageId);
        log.info("[MemoryDataCenter] 未确认消息从确认队列中移除! queueName={}, messageId={}", queueName, messageId);
    }

    public Message getMessageWaitACK(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageWaitACKMap = queueMessageMapWaitACK.get(queueName);
        if(messageWaitACKMap == null){
            return null;
        }
        return messageWaitACKMap.get(messageId);
    }

    /**
     * 重启系统后将硬盘上的持久化数据加载到内存中
     * 恢复所有的交换机
     * 恢复所有的队列
     * 恢复所有的绑定关系
     * 恢复所有的消息
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, ClassNotFoundException, MqException {
        //恢复所有的交换机
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            insertExchange(exchange);
        }
        //恢复所有的队列
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            insertQueue(queue);
        }
        //恢复所有的绑定关系
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //恢复所有的消息
        for(MSGQueue queue : queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages) {
                messagesMap.put(message.getMessageID(),message);
            }
        }
    }
}
